package com.tbvtc.agriculturalcreditchainserver2025.ws;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.tbvtc.agriculturalcreditchainserver2025.entity.ChatMessage;
import com.tbvtc.agriculturalcreditchainserver2025.entity.User; // 替换为你的User实体类
import com.tbvtc.agriculturalcreditchainserver2025.entity.websocket.WebSocketMessage;
import com.tbvtc.agriculturalcreditchainserver2025.exception.FoodException;
import com.tbvtc.agriculturalcreditchainserver2025.login.LoginUser;
import com.tbvtc.agriculturalcreditchainserver2025.login.LoginUserHolder;
import com.tbvtc.agriculturalcreditchainserver2025.service.IChatMessageService;
import com.tbvtc.agriculturalcreditchainserver2025.service.IUserService; // 需注入用户查询服务
import com.tbvtc.agriculturalcreditchainserver2025.utils.SpringContextUtil;
import com.tbvtc.agriculturalcreditchainserver2025.vo.chatMessage.GetAllChatValueVo;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.fisco.bcos.web3j.tx.txdecode.BaseException;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@ServerEndpoint(value = "/chatWs/{id}")
@Tag(name = "websocket服务端")
public class WebSocketServer {
    private static int onlineCount = 0;
    private static final ConcurrentHashMap<String, WebSocketServer> webSocketSet = new ConcurrentHashMap<>();
    // 新增：用户ID -> 登录用户信息映射（线程安全）
    private static final ConcurrentHashMap<String, User> userIdToUserMap = new ConcurrentHashMap<>();
    // 新增：线程池（核心线程5，最大10，队列容量50，空闲线程60秒回收）
    private static final ExecutorService messageThreadPool = new ThreadPoolExecutor(
            5, 10, 60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(50),
            new ThreadFactory() {
                private final AtomicInteger count = new AtomicInteger(0);
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "ws-message-thread-" + count.incrementAndGet());
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时回退到调用线程执行，避免任务丢失
    );

    private final Object sendLock = new Object();
    private volatile boolean isSending = false;
    private Session session;
    private static final Logger log = LogManager.getLogger(WebSocketServer.class);
    private String id = ""; // 连接URL中的用户ID（与登录用户ID一致）
    IChatMessageService chatMessageService = SpringContextUtil.getBean(IChatMessageService.class);
    // 新增：注入用户服务，用于通过ID查询登录用户信息
    IUserService userService = SpringContextUtil.getBean(IUserService.class);

    /**
     * 连接建立成功调用的方法：绑定用户身份
     */
    @OnOpen
    public void onOpen(@PathParam(value = "id") String id, Session session) {
        this.session = session;
        this.id = id;
        webSocketSet.put(id, this);
        addOnlineCount();

        // 新增：查询登录用户信息并绑定（假设id是用户唯一标识，需确保客户端传递的是登录后的用户ID）
        User loginUser = userService.getById(id); // 替换为你的用户查询逻辑（如从缓存/数据库查询）
        if (loginUser == null) {
            log.error("用户ID:{} 未登录或不存在，WebSocket连接建立失败", id);
            try {
                session.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, "用户未登录"));
            } catch (IOException e) {
                log.error("关闭未登录用户连接失败", e);
            }
            return;
        }
        userIdToUserMap.put(id, loginUser); // 绑定用户ID与用户信息
        log.info("用户{}（用户名：{}）加入！当前在线人数为{}", id, loginUser.getUsername(), getOnlineCount());
    }

    /**
     * 连接关闭：清理资源
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this.id);
        userIdToUserMap.remove(this.id); // 移除用户映射
        subOnlineCount();
        LoginUserHolder.clear(); // 清除当前线程的用户信息（防止线程池复用）
        log.info("用户{}连接关闭！当前在线人数为{}", this.id, getOnlineCount());
    }

    /**
     * 收到客户端消息：提交到线程池异步处理
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("来自客户端{}的消息:{}", this.id, message);
        // 1. 校验用户是否已绑定
        User loginUser = userIdToUserMap.get(this.id);
        if (loginUser == null) {
            log.error("用户{}未绑定登录信息，拒绝处理消息", this.id);
            try {
                sendMessage("用户未登录，消息发送失败");
            } catch (IOException e) {
                log.error("回复未登录提示失败", e);
            }
            return;
        }

        // 2. 提交到线程池异步处理（避免阻塞IO线程）
        messageThreadPool.submit(() -> {
            try {
                // 3. 绑定用户信息到当前线程（供线程池内业务逻辑获取）
                LoginUserHolder.setLoginUser(new LoginUser(loginUser.getPhone(), (long)loginUser.getId()));
                // 4. 解析消息并处理
                WebSocketMessage webSocketMessage = JSON.parseObject(message, WebSocketMessage.class);
                Long chatId = chatMessageService.singleSend(webSocketMessage);
                // 5. 按消息类型分发处理
                switch (webSocketMessage.getChatType()) {
                    case BROADCAST_SEND:
                        broadcastSend(webSocketMessage);
                        break;
                    case SINGLE_SEND:
                        singleSend(webSocketMessage, chatId);
                        break;
                    case GROUP_SEND:
                        groupSend(webSocketMessage);
                        break;
                    default:
                        throw new FoodException(0, "消息发送失败");
                }
            } catch (Exception e) {
                log.error("线程池处理消息失败", e);
                try {
                    sendMessage("消息处理失败：" + e.getMessage());
                } catch (IOException ex) {
                    log.error("回复消息处理失败提示失败", ex);
                }
            } finally {
                // 6. 清除线程局部变量（必须执行，避免内存泄漏）
                LoginUserHolder.clear();
            }
        });
    }

    /**
     * 群发消息（无需修改核心逻辑，线程池内已能通过UserContextHolder获取当前用户）
     */
    private void groupSend(WebSocketMessage webSocketMessage) {
        // 如需获取当前发送用户，直接调用：User currentUser = UserContextHolder.getUser();
        for (String receiverId : webSocketMessage.getReceiverId()) {
            if (webSocketMessage.getFilePath().isEmpty()) {
                webSocketMessage.getContent().forEach(content -> {
                    WebSocketMessage socketMessage = new WebSocketMessage();
                    BeanUtil.copyProperties(webSocketMessage, socketMessage);
                    socketMessage.setContent(List.of(content));
                    String jsonString = JSON.toJSONString(socketMessage);
                    try {
                        sendToUser(jsonString, receiverId);
                    } catch (IOException e) {
                        throw new FoodException(0, "消息发送失败");
                    }
                });
            } else {
                webSocketMessage.getFilePath().forEach(filePath -> {
                    WebSocketMessage socketMessage = new WebSocketMessage();
                    BeanUtil.copyProperties(webSocketMessage, socketMessage);
                    socketMessage.setFilePath(List.of(filePath));
                    webSocketMessage.getMetadata().forEach(metadata -> {
                        if (Objects.equals(metadata.getUrl(), filePath)) {
                            socketMessage.setMetadata(List.of(metadata));
                        }
                    });
                    String jsonString = JSON.toJSONString(socketMessage);
                    try {
                        sendToUser(jsonString, receiverId);
                    } catch (IOException e) {
                        throw new FoodException(0, "消息发送失败");
                    }
                });
            }
        }
        log.info("用户{}群发消息成功", LoginUserHolder.getLoginUser().getUserId());
    }

    /**
     * 单聊消息（线程池内可获取当前发送用户）
     */
    private void singleSend(WebSocketMessage webSocketMessage, Long chatId) {
        LoginUser loginUser = LoginUserHolder.getLoginUser();// 获取当前登录用户

        if (webSocketMessage.getFilePath().isEmpty()){
            webSocketMessage.getContent().forEach(content -> {
                webSocketMessage.setContent(List.of(content));
                String jsonString = JSON.toJSONString(webSocketMessage);
                try {
                    sendToUserContent( webSocketMessage.getReceiverId().get(0), chatId);
                } catch (IOException e) {
                    throw new FoodException(0, "消息发送失败");
                }
            });
        } else {
            webSocketMessage.getFilePath().forEach(filePath -> {
                WebSocketMessage socketMessage = new WebSocketMessage();
                BeanUtil.copyProperties(webSocketMessage, socketMessage);
                socketMessage.setFilePath(List.of(filePath));
                webSocketMessage.getMetadata().forEach(metadata -> {
                    if (Objects.equals(metadata.getUrl(), filePath)) {
                        socketMessage.setMetadata(List.of(metadata));
                    }
                });
                String jsonString = JSON.toJSONString(socketMessage);
                try {
                    sendToUserContent( webSocketMessage.getReceiverId().get(0),chatId);
                } catch (IOException e) {
                    throw new FoodException(0, "消息发送失败");
                }
            });
        }
        log.info("用户{}单聊消息发送成功", loginUser.getUserId());
    }

    /**
     * 广播消息（线程池内可获取当前发送用户）
     */
    private void broadcastSend(WebSocketMessage webSocketMessage) throws FoodException {
        LoginUser loginUser = LoginUserHolder.getLoginUser();// 获取当前登录用户

        if (webSocketMessage.getFilePath().isEmpty()) {
            webSocketMessage.getContent().forEach(content -> {
                WebSocketMessage socketMessage = new WebSocketMessage();
                BeanUtil.copyProperties(webSocketMessage, socketMessage);
                socketMessage.setContent(List.of(content));
                String jsonString = JSON.toJSONString(socketMessage);
                sendToAllClient(jsonString);
            });
        } else {
            webSocketMessage.getFilePath().forEach(filePath -> {
                WebSocketMessage socketMessage = new WebSocketMessage();
                BeanUtil.copyProperties(webSocketMessage, socketMessage);
                socketMessage.setFilePath(List.of(filePath));
                webSocketMessage.getMetadata().forEach(metadata -> {
                    if (Objects.equals(metadata.getUrl(), filePath)) {
                        socketMessage.setMetadata(List.of(metadata));
                    }
                });
                String jsonString = JSON.toJSONString(socketMessage);
                sendToAllClient(jsonString);
            });
        }
        log.info("用户{}广播消息成功", loginUser.getUserId());
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用户{}的WebSocket连接发生错误", this.id);
        error.printStackTrace();
        LoginUserHolder.clear(); // 异常时清除线程局部变量
    }

    // 以下方法（sendMessage、sendToUser、sendToUserContent、sendToAllClient、getOnlineCount等）保持不变
    public boolean sendMessage(String message) throws IOException {
        if (this.session == null || !this.session.isOpen()) {
            log.warn("客户端 [{}] 的 WebSocket Session 已关闭，跳过消息发送", this.id);
            return false;
        }
        synchronized (sendLock) {
            if (isSending) {
                log.warn("客户端 [{}] 正在发送消息，本次消息等待重试", this.id);
                return false;
            }
            try {
                isSending = true;
                this.session.getBasicRemote().sendText(message);
                log.debug("向客户端 [{}] 发送消息：{}", this.id, message);
                return true;
            } finally {
                isSending = false;
            }
        }
    }
    public void sendToUser(String message, String sendUserId) throws IOException {
        if (webSocketSet.get(sendUserId) != null) {
            if (!id.equals(sendUserId)) {
                webSocketSet.get(sendUserId).sendMessage("用户" + id + "发来消息：" + message);
            } else {
                webSocketSet.get(sendUserId).sendMessage(message);
            }
        } else {
            log.warn("用户{}不在线，无法发送消息", sendUserId);
        }
    }

    public void sendToUserContent( String sendUserId, Long chatId) throws IOException {
        String count = chatMessageService.getCount( sendUserId, chatId);
        if (webSocketSet.get(sendUserId) != null) {
            if (!id.equals(sendUserId)) {
                webSocketSet.get(sendUserId).sendMessage("用户" + id + "发来消息：" + count);
                webSocketSet.get(id).sendMessage(count);
            } else {
                webSocketSet.get(sendUserId).sendMessage(count);
                webSocketSet.get(id).sendMessage(count);
            }
        } else {
            webSocketSet.get(id).sendMessage("用户" + sendUserId + "不在线，消息已保存");
        }
    }

    public void sendToAllClient(String message) throws FoodException {
        if (getOnlineCount() == 0) {
            return;
        }
        for (String key : webSocketSet.keySet()) {
            try {
                webSocketSet.get(key).sendMessage(message);
            } catch (IOException e) {
                throw new FoodException(0, "消息发送失败");
            }
        }
    }
    /**
     * 静态方法：向指定用户发送消息
     */
    public static void sendToUserStatic(String message, String userId) throws IOException {
        WebSocketServer client = webSocketSet.get(userId);
        if (client != null && client.session.isOpen()) {
            client.sendMessage(message);
        } else {
            log.warn("用户[{}]不在线，无法发送消息", userId);
        }
    }


    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }

    // 新增：销毁线程池（避免应用关闭时资源泄漏）
    @PreDestroy
    public void destroyThreadPool() {
        log.info("开始关闭WebSocket消息处理线程池");
        messageThreadPool.shutdown();
        try {
            if (!messageThreadPool.awaitTermination(30, TimeUnit.SECONDS)) {
                messageThreadPool.shutdownNow();
            }
        } catch (InterruptedException e) {
            messageThreadPool.shutdownNow();
        }
        log.info("WebSocket消息处理线程池已关闭");
    }
}